package com.lpssfxy.offline.related

import com.lpssfxy.offline.utils.AppUtils
import com.lpssfxy.offline.utils.AppUtils.{MONGODB_RELATED_RECS_COLLECTION, loadProductData, saveRecommendationsToMongoDB}
import org.apache.spark.sql.{functions => F}

/**
 * 为了验证推荐结果的准确度（商品维度）
 * 这里将通过计算商品的Jaccard相似度，得到productId和对应的真实相关商品列表，存储与mongo中。
 * 将其作为验证离线推荐结果中商品相似度准确率的标准
 */
object ProductRelatedByTags {

  def main(args: Array[String]): Unit = {
    // 创建一个SparkSession对象，用于与Spark集群进行交互
    // 这里指定应用程序名称为"ProductRelatedByTags"
    val spark = AppUtils.createSparkSession("ProductRelatedByTags",AppUtils.getSparkCores)

    // 调用AppUtils工具类中的loadProductData方法从数据源（如MongoDB）加载商品数据
    // 加载后的数据存储在一个DataFrame中，每一行代表一个商品的信息
    val productDF = loadProductData(spark)

    // 使用Spark SQL的functions（导入为F）中的split函数，将productDF中的tags列按照竖线"|"进行分割
    // 分割后的结果存储在一个新的数组列"tagArray"中，最终得到一个新的DataFrame
    val productWithTagsArrayDF = productDF.withColumn("tagArray", F.split(F.col("tags"), "\\|"))

    // 自定义一个用户定义函数（UDF），用于计算两个标签集合的Jaccard相似度
    // Jaccard相似度是一种衡量两个集合相似程度的指标，计算公式为：交集元素个数 / 并集元素个数
    val jaccardUDF = F.udf((tags1: Seq[String], tags2: Seq[String]) => {
      // 将输入的两个标签序列转换为集合，方便后续计算交集和并集
      val set1 = tags1.toSet
      val set2 = tags2.toSet
      // 计算两个集合的交集元素个数
      val intersection = set1.intersect(set2).size
      // 计算两个集合的并集元素个数
      val union = set1.union(set2).size
      // 为了避免除零错误，如果并集元素个数为0，则返回相似度为0.0
      // 否则，返回交集元素个数除以并集元素个数的结果
      if (union == 0) 0.0 else intersection.toDouble / union.toDouble
    })

    // 计算所有商品对的相似度，生成一个新的DataFrame
    val productPairsDF = productWithTagsArrayDF.as("p1")
      // 使用crossJoin方法将DataFrame自身进行交叉连接，得到所有可能的商品对
      .crossJoin(productWithTagsArrayDF.as("p2"))
      // 过滤掉商品对中两个商品ID相同的情况，避免自身与自身对比
      .filter(F.col("p1.productId") =!= F.col("p2.productId"))
      // 选择需要的列，包括第一个商品的ID、第二个商品的ID以及它们的Jaccard相似度
      .select(
        F.col("p1.productId").as("productId"),
        F.col("p2.productId").as("relatedProductId"),
        // 调用前面定义的jaccardUDF函数计算两个商品标签集合的相似度
        jaccardUDF(F.col("p1.tagArray"), F.col("p2.tagArray")).as("similarity")
      )

    // 设置一个相似度阈值，用于筛选出相似度较高的商品对
    // 这里假设阈值为0.3，只有相似度大于0.3的商品对才会被保留
    val threshold = 0.3
    // 对productPairsDF进行处理，得到每个商品对应的相关商品列表
    val relatedProductsDF = productPairsDF
      // 过滤掉相似度小于等于阈值的商品对
      .filter(F.col("similarity") > threshold)
      // 按照productId进行分组，将每个商品的相关商品信息聚合在一起
      .groupBy("productId")
      // 使用agg函数进行聚合操作，将符合条件的相关商品信息收集到一个列表中
      .agg(F.collect_list(F.struct(
        // 将相关商品的ID转换为整数类型，并命名为productId
        F.col("relatedProductId").cast("int").as("productId"),
        // 将相似度命名为score
        F.col("similarity").as("score")
      )).as("recs"))

    // 调用AppUtils工具类中的saveRecommendationsToMongoDB方法
    // 将处理好的relatedProductsDF数据保存到MongoDB的relatedProductSimList集合中
    saveRecommendationsToMongoDB(relatedProductsDF, MONGODB_RELATED_RECS_COLLECTION)

    // 停止SparkSession，释放资源
    spark.stop()
  }
}